package source.javadog.mysource;


import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import static source.javadog.mysource.MyDataSources.*;

/**
 * Author : manyu
 * Date: 2018/9/9 22:36
 * Description:
 */
public class MyPoolJdbcSource extends MyPoolSource<Connection> {

    protected final ConnectionPoolDataSource source;

    protected final ArrayBlockingQueue<PooledConnection> queue;

    protected final ConnectionEventListener listener;

    public MyPoolJdbcSource(ArrayBlockingQueue queue, Semaphore semaphore, Map<String, String> dbMap, Logger logger) {
        super(semaphore, dbMap, logger);
        this.source = createDataSource(dbMap);
        this.queue = queue;
        this.listener = new ConnectionEventListener() {

            @Override
            public void connectionClosed(ConnectionEvent event) {
                PooledConnection pc = (PooledConnection) event.getSource();
                if (queue.offer(pc)) {
                    saveCounter.incrementAndGet();
                } else {
                    try {
                        pc.close();
                    } catch (Exception e) {
                        logger.log(Level.INFO, MyDataSource.class.getSimpleName() + " " + pc + " close error", e);
                    }
                }
            }

            @Override
            public void connectionErrorOccurred(ConnectionEvent event) {
                usingCounter.decrementAndGet();
                if ("08S01".equals(event.getSQLException().getSQLState())) return; //MySQL特性， 长时间连接没使用会抛出com.mysql.jdbc.exceptions.jdbc4.CommunicationsException
                logger.log(Level.WARNING, "connectionErronOccurred  [" + event.getSQLException().getSQLState() + "]", event.getSQLException());
            }
        };
    }

    private ConnectionPoolDataSource createDataSource(Map<String, String> dbMap) {
        try {
            return createDataSource(dbMap.get(JDBC_DRIVER), dbMap.get(JDBC_URL), dbMap.get(JDBC_USER), dbMap.get(JDBC_PWD));
        } catch (Exception ex) {
            throw new RuntimeException("(" + dbMap + ") have no jdbc parameters", ex);
        }
    }

    private ConnectionPoolDataSource createDataSource(String source0, String url, String user, String password) throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
        String source;
        try {
            Thread.currentThread().getContextClassLoader().loadClass("com.mysql.cj.jdbc.MysqlConnectionPoolDataSource");
            source = "com.mysql.cj.jdbc.MysqlConnectionPoolDataSource";
        } catch (Throwable e) {
            source = "com.mysql.jdbc.jdbc2.optional.MysqlConnectionPoolDataSource";
        }
        final Class clazz = Thread.currentThread().getContextClassLoader().loadClass(source);
        Object pdsource = clazz.getDeclaredConstructor().newInstance();
        Method seturlm;
        try {
            seturlm = clazz.getMethod("setUrl", String.class);
        } catch (Exception e) {
            seturlm = clazz.getMethod("setURL", String.class);
        }
        seturlm.invoke(pdsource, url);
        clazz.getMethod("setUser", String.class).invoke(pdsource, user);
        clazz.getMethod("setPassword", String.class).invoke(pdsource, password);
        return (ConnectionPoolDataSource) pdsource;
    }

    @Override
    protected int getDefaultPort() {
        return 0;
    }

    @Override
    public Connection poll() {
        return poll(0, null);
    }

    @Override
    public void offerConnection(final Connection conn) {
        if (conn == null) return;
        try {
            conn.close();
        } catch (Exception e) {
            logger.log(Level.WARNING, "closeSQLConnection abort", e);
        }
    }

    private Connection poll(final int count, SQLException e) {
        if (count >= 3) {
            logger.log(Level.WARNING, "create pooled connection error", e);
            throw new RuntimeException(e);
        }
        PooledConnection result = queue.poll();
        if (result == null) {
            if (usingCounter.get() >= maxconns) {
                try {
                    result = queue.poll(6, TimeUnit.SECONDS);
                } catch (Exception t) {
                    logger.log(Level.WARNING, "take pooled connection error", t);
                }
            }
            if (result == null) {
                try {
                    result = source.getPooledConnection();
                    result.addConnectionEventListener(listener);
                    usingCounter.incrementAndGet();
                } catch (SQLException ex) {
                    return poll(count + 1, ex);
                }
                creatCounter.incrementAndGet();
            }
        } else {
            cycleCounter.incrementAndGet();
        }
        Connection conn;
        try {
            conn = result.getConnection();
            if (!conn.isValid(1)) {
                logger.info("sql connection is not vaild");
                usingCounter.decrementAndGet();
                return poll(0, null);
            }
        } catch (SQLException ex) {
            if (!"08S01".equals(ex.getSQLState())) {//MySQL特性， 长时间连接没使用会抛出com.mysql.jdbc.exceptions.jdbc4.CommunicationsException
                logger.log(Level.FINER, "result.getConnection from pooled connection abort [" + ex.getSQLState() + "]", ex);
            }
            return poll(0, null);
        }
        return conn;
    }
}
